package com.tang.framework.delay.support;

import com.alibaba.fastjson2.JSON;
import com.tang.common.constant.delay.DelayConstant;
import com.tang.framework.delay.support.domain.DelayMethod;
import com.tang.framework.delay.support.domain.Message;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RBucket;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.http.MediaType;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import jakarta.annotation.Resource;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author tang jun liang
 * @Date 2023/4/19 14:57
 */
@Slf4j
public class DelayMethodHandler {

    private final DelayMethod delayMethod;

    private final RDelayedQueue<Message<?>> delayedQueue;

    private final RBlockingQueue<Message<?>> finalQueue;

    private final RedissonClient redissonClient;

    private final Method method;

    private final Object bean;

    private final DelayMethod.Config config;

    private final DataSourceTransactionManager dataSourceTransactionManager;

    private final static String FINAL_QUEUE_KEY = "FINAL_QUEUE";

    private final TransactionDefinition transactionDefinition;

    public DelayMethodHandler(DelayMethod delayMethod,
                              DataSourceTransactionManager dataSourceTransactionManager,
                              TransactionDefinition transactionDefinition,
                              RedissonClient redissonClient
    ) {
        this.delayMethod = delayMethod;
        this.dataSourceTransactionManager = dataSourceTransactionManager;
        this.transactionDefinition = transactionDefinition;
        this.redissonClient = redissonClient;
        this.method = delayMethod.getTargetMethod();
        RBlockingQueue<Message<?>> blockingQueue = this.redissonClient.getBlockingQueue(delayMethod.getProperties().getName());
        this.delayedQueue = this.redissonClient.getDelayedQueue(blockingQueue);
        this.config = delayMethod.getProperties();
        this.finalQueue = redissonClient.getBlockingQueue(DelayConstant.FINAL_QUEUE_KEY);
        this.bean = delayMethod.getBean();
    }

    @SuppressWarnings("all")
    public void handler(Message<?> message, Object... args) {
        if (checkExecuted(message)) {
            return;
        }
        TransactionStatus transaction = null;
        if (config.getEnableTransaction()){
            transaction = dataSourceTransactionManager.getTransaction(transactionDefinition);
        }
        try {
            log.debug("开始执行队列任务: {}, 方法名{}, 消息id:{}, 参数内容:{}",
                    delayMethod.getProperties().getName(),
                    method.getName(), message.getId(), JSON.toJSONString(message));
            method.setAccessible(true);
            if (method.getParameters()[0].equals(Message.class)) {
                method.invoke(bean, message);
            } else {
                method.invoke(bean, message.getBody());
            }
            setRunTip(message);
            if (config.getEnableTransaction()){
                dataSourceTransactionManager.commit(transaction);
            }
        } catch (Exception e) {
            if (config.getEnableTransaction()){
                dataSourceTransactionManager.rollback(transaction);
            }
            message.increment();
            log.error("执行{}队列任务失败: 方法名{}, 消息id:{}, 参数内容:{}",
                    delayMethod.getProperties().getName(),
                    method.getName(), message.getId(), JSON.toJSONString(message));
            //超出最大尝试次数,把信息放到最终队列去 不影响正常的消息消费
            if (message.getRetry() > config.getRetry()) {
                log.debug("队列任务{}超出重试次数，已将任务投放到最终队列", message.getId());
                //记录一下是哪个延迟方法要执行
                message.setMethodInfo(createMethodInfo());
                finalQueue.offer(message);
            }
            delayedQueue.offer(message, config.getRetryIntervalTime(), TimeUnit.SECONDS);
        }
    }


    private Boolean checkExecuted(Message<?> message) {
        RBucket<String> checkRun = redissonClient.getBucket("delay:" + message.getId());
        return checkRun.isExists();
    }

    private void setRunTip(Message<?> message) {
        RBucket<String> runTip = redissonClient.getBucket("delay:" + message.getId());
        runTip.set("run");
        runTip.expire(Duration.ofMinutes(1L));
    }

    private Message.MethodInfo createMethodInfo() {
        Message.MethodInfo methodInfo = new Message.MethodInfo();
        methodInfo.setMethodName(method.getName());
        methodInfo.setQueueName(config.getName());
        methodInfo.setClazz(delayMethod.getClass());
        return methodInfo;
    }


}
